package net.bwie.realtime.jtp.dwd.log.job;

import com.alibaba.fastjson.JSON;
import net.bwie.realtime.jtp.common.utils.KafkaUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.time.LocalDate;
import java.util.HashMap;
import java.util.Map;

/**
 * 简介说明: 达摩盘用户画像标签处理
 */
public class UserProfileTagsWeightJob {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 从启动参数获取配置
        ParameterTool params = ParameterTool.fromArgs(args);
        env.getConfig().setGlobalJobParameters(params);

        env.setParallelism(params.getInt("parallelism", 1));

        DataStream<String> user_profile_DataStream = KafkaUtil.consumerKafka(env, "user_profile_tags");

        // 清理日志并处理
        DataStream<UserProfile> processedStream = processUserProfiles(user_profile_DataStream);
//        processedStream.print();
        // 将处理结果发送到输出主题
        SingleOutputStreamOperator<String> map = processedStream.map(user -> JSON.toJSONString(user));

        KafkaUtil.producerKafka(map,"user_profile_Age_tag");

        // 执行名称包含时间戳，方便查找
        env.execute("UserProfileTagsJob-" + System.currentTimeMillis());
    }

    /**
     * 用户画像处理流程
     */
    private static DataStream<UserProfile> processUserProfiles(DataStream<String> userProfileDataStream) {
        // 1. 日志清洗
        DataStream<String> cleanedStream = logCleaned(userProfileDataStream);

        // 2. 转换JSON字符串为UserProfile对象
        DataStream<UserProfile> parsedStream = cleanedStream.map(new MapFunction<String, UserProfile>() {
            @Override
            public UserProfile map(String json) throws Exception {
                return JSON.parseObject(json, UserProfile.class);
            }
        });

        // 3. 应用年龄标签处理
        DataStream<UserProfile> processedStream = parsedStream.map(new AgeLabelProcessor());

        return processedStream;
    }

    /**
     * 日志清洗：过滤无效JSON数据
     */
    private static DataStream<String> logCleaned(DataStream<String> userProfileDataStream) {
        OutputTag<String> dirtyTag = new OutputTag<String>("dirty_log"){};

        SingleOutputStreamOperator<String> cleanedStream = userProfileDataStream.process(
                new ProcessFunction<String, String>() {
                    @Override
                    public void processElement(String value, Context ctx, Collector<String> out) throws Exception {
                        try {
                            JSON.parseObject(value);
                            out.collect(value);
                        } catch (Exception e) {
                            ctx.output(dirtyTag, value);
                        }
                    }
                }
        );

        // 处理脏数据
        DataStream<String> dirtyStream = cleanedStream.getSideOutput(dirtyTag);
        KafkaUtil.producerKafka(dirtyStream, "user_profile_dirty_tags");

        return cleanedStream;
    }

    /**
     * 用户画像数据结构
     */
    public static class UserProfile {
        private int user_id;
        private Integer birth_year;
        private String gender_label;
        private double weight;
        private double height;
        private String constellation;
        private String category_pref;
        private String brand_pref;
        private double price_sensitivity;
        private TimeBehavior time_behavior;
        private String search_terms;
        private Integer social_interaction;
        private String device_info;
        private String age_label; // 计算后新增的字段

        // Getters and Setters
        public int getUser_id() { return user_id; }
        public void setUser_id(int user_id) { this.user_id = user_id; }

        public Integer getBirth_year() { return birth_year; }
        public void setBirth_year(Integer birth_year) { this.birth_year = birth_year; }

        public String getGender_label() { return gender_label; }
        public void setGender_label(String gender_label) { this.gender_label = gender_label; }

        public double getWeight() { return weight; }
        public void setWeight(double weight) { this.weight = weight; }

        public double getHeight() { return height; }
        public void setHeight(double height) { this.height = height; }

        public String getConstellation() { return constellation; }
        public void setConstellation(String constellation) { this.constellation = constellation; }

        public String getCategory_pref() { return category_pref; }
        public void setCategory_pref(String category_pref) { this.category_pref = category_pref; }

        public String getBrand_pref() { return brand_pref; }
        public void setBrand_pref(String brand_pref) { this.brand_pref = brand_pref; }

        public double getPrice_sensitivity() { return price_sensitivity; }
        public void setPrice_sensitivity(double price_sensitivity) { this.price_sensitivity = price_sensitivity; }

        public TimeBehavior getTime_behavior() { return time_behavior; }
        public void setTime_behavior(TimeBehavior time_behavior) { this.time_behavior = time_behavior; }

        public String getSearch_terms() { return search_terms; }
        public void setSearch_terms(String search_terms) { this.search_terms = search_terms; }

        public Integer getSocial_interaction() { return social_interaction; }
        public void setSocial_interaction(Integer social_interaction) { this.social_interaction = social_interaction; }

        public String getDevice_info() { return device_info; }
        public void setDevice_info(String device_info) { this.device_info = device_info; }

        public String getAge_label() { return age_label; }
        public void setAge_label(String age_label) { this.age_label = age_label; }
    }

    /**
     * 时间行为数据结构
     */
    public static class TimeBehavior {
        private Integer morning;
        private Integer afternoon;
        private Integer night;

        public Integer getMorning() { return morning; }
        public void setMorning(Integer morning) { this.morning = morning; }

        public Integer getAfternoon() { return afternoon; }
        public void setAfternoon(Integer afternoon) { this.afternoon = afternoon; }

        public Integer getNight() { return night; }
        public void setNight(Integer night) { this.night = night; }
    }

    /**
     * 年龄标签处理器
     */
    public static class AgeLabelProcessor extends RichMapFunction<UserProfile, UserProfile> {
        // 年龄分段定义
        private static final String AGE_18_24 = "18-24岁";
        private static final String AGE_25_29 = "25-29岁";
        private static final String AGE_30_34 = "30-34岁";
        private static final String AGE_35_39 = "35-39岁";
        private static final String AGE_40_49 = "40-49岁";
        private static final String AGE_50_PLUS = "50岁以上";

        // 维度权重
        private static final double CATEGORY_WEIGHT = 0.30;
        private static final double BRAND_WEIGHT = 0.20;
        private static final double PRICE_SENSITIVITY_WEIGHT = 0.15;
        private static final double TIME_BEHAVIOR_WEIGHT = 0.10;
        private static final double SEARCH_TERMS_WEIGHT = 0.10;
        private static final double SOCIAL_INTERACTION_WEIGHT = 0.10;
        private static final double DEVICE_WEIGHT = 0.05;

        // 预定义评分规则
        private Map<String, Integer> categoryScores;
        private Map<String, Integer> brandScores;
        private Map<String, Integer> searchTermScores;

        @Override
        public void open(Configuration parameters) {
            // 初始化评分规则
            initScoringRules();
        }

        @Override
        public UserProfile map(UserProfile profile) throws Exception {
            // 优先使用出生年份直接计算
            if (isValidBirthYear(profile.getBirth_year())) {
                int age = calculateAge(profile.getBirth_year());
                profile.setAge_label(getAgeBucketByAge(age));
            }
            // 多维度加权计算
            else {
                double score = calculateWeightedAgeScore(profile);
                profile.setAge_label(getAgeBucketByScore(score));
            }
            return profile;
        }

        private void initScoringRules() {
            // 类目偏好评分规则
            categoryScores = new HashMap<>();
            categoryScores.put("数码电子", 85);
            categoryScores.put("美妆护肤", 80);
            categoryScores.put("运动户外", 75);
            categoryScores.put("母婴用品", 45);
            categoryScores.put("保健品", 40);
            categoryScores.put("金融理财", 35);
            categoryScores.put("退休规划", 20);

            // 品牌偏好评分规则
            brandScores = new HashMap<>();
            brandScores.put("设计师品牌", 90);
            brandScores.put("潮牌", 85);
            brandScores.put("奢侈品牌", 70);
            brandScores.put("大众品牌", 50);
            brandScores.put("经典品牌", 40);

            // 搜索词评分规则
            searchTermScores = new HashMap<>();
            searchTermScores.put("电竞", 90);
            searchTermScores.put("追星", 85);
            searchTermScores.put("抗初老", 80);
            searchTermScores.put("轻奢", 75);
            searchTermScores.put("育儿经验", 45);
            searchTermScores.put("退休规划", 30);
        }

        private boolean isValidBirthYear(Integer birthYear) {
            if (birthYear == null) return false;
            int currentYear = LocalDate.now().getYear();
            return birthYear > 1900 && birthYear < currentYear;
        }

        private int calculateAge(int birthYear) {
            return LocalDate.now().getYear() - birthYear;
        }

        private String getAgeBucketByAge(int age) {
            if (age < 18) return AGE_18_24; // 未成年处理
            if (age <= 24) return AGE_18_24;
            if (age <= 29) return AGE_25_29;
            if (age <= 34) return AGE_30_34;
            if (age <= 39) return AGE_35_39;
            if (age <= 49) return AGE_40_49;
            return AGE_50_PLUS;
        }

        private double calculateWeightedAgeScore(UserProfile profile) {

            double totalScore = 0;

            // 1. 类目偏好维度 (30%)
            String category = profile.getCategory_pref();
            int categoryScore = categoryScores.getOrDefault(category, 50);
            totalScore += categoryScore * CATEGORY_WEIGHT;

            // 2. 品牌偏好维度 (20%)
            String brand = profile.getBrand_pref();
            int brandScore = brandScores.getOrDefault(brand, 50);
            totalScore += brandScore * BRAND_WEIGHT;

            // 3. 价格敏感度维度 (15%)
            totalScore += (profile.getPrice_sensitivity() * 100) * PRICE_SENSITIVITY_WEIGHT;

            // 4. 时间行为维度 (10%)
            TimeBehavior time = profile.getTime_behavior();
            if (time != null && time.getNight() != null) {
                totalScore += Math.min(time.getNight() * 7, 100) * TIME_BEHAVIOR_WEIGHT;
            } else {
                totalScore += 50 * TIME_BEHAVIOR_WEIGHT;
            }

            // 5. 搜索词分析维度 (10%)
            int searchScore = calculateSearchTermScore(profile.getSearch_terms());
            totalScore += searchScore * SEARCH_TERMS_WEIGHT;

            // 6. 社交互动行为 (10%)
            Integer social = profile.getSocial_interaction();
            if (social != null) {
                totalScore += Math.min(social * 2, 100) * SOCIAL_INTERACTION_WEIGHT;
            } else {
                totalScore += 50 * SOCIAL_INTERACTION_WEIGHT;
            }

            // 7. 设备信息维度 (5%)
            totalScore += getDeviceScore(profile.getDevice_info()) * DEVICE_WEIGHT;

            return totalScore;
        }

        private int calculateSearchTermScore(String searchTerms) {
            if (StringUtils.isBlank(searchTerms)) return 50;

            int total = 0;
            int count = 0;
            String[] terms = searchTerms.split(" ");

            for (String term : terms) {
                Integer score = searchTermScores.get(term);
                if (score != null) {
                    total += score;
                    count++;
                }
            }

            return count > 0 ? total / count : 50;
        }

        private int getDeviceScore(String deviceInfo) {
            if (StringUtils.isBlank(deviceInfo)) return 55;

            String device = deviceInfo.toLowerCase();
            if (device.contains("iphone")) return 85;
            if (device.contains("pro")) return 80;
            if (device.contains("安卓旗舰")) return 75;
            if (device.contains("小米")) return 65;
            return 55; // 默认分数
        }

        private String getAgeBucketByScore(double score) {
            if (score >= 85) return AGE_18_24;
            if (score >= 70) return AGE_25_29;
            if (score >= 55) return AGE_30_34;
            if (score >= 45) return AGE_35_39;
            if (score >= 30) return AGE_40_49;
            return AGE_50_PLUS;
        }
    }
}